package com.xzx.flink.tableapi;

import com.xzx.flink.bean.ClickEvent;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/21 17:47
 */
public class TableAPI_01_Demo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<ClickEvent> source = env.fromElements(
                new ClickEvent("Alice", "./home", 1000L),
                new ClickEvent("Bob", "./cart", 1000L),
                new ClickEvent("Alice", "./prod?id=1", 5 * 1000L),
                new ClickEvent("Cary", "./home", 60 * 1000L),
                new ClickEvent("Bob", "./prod?id=3", 90 * 1000L),
                new ClickEvent("Alice", "./prod?id=7", 105 * 1000L)
        );
        //表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 将数据流转换成表, Table 如果没有指定tableName的话默认是UnnamedTable$0这种风格，第一个表$0,第二个表$1,会议依次递增
        Table eventTable = tableEnv.fromDataStream(source);
        //执行sql的方式提取数据
        Table resultTable = tableEnv.sqlQuery("select url, user from " + eventTable);
        //输出
        tableEnv.toDataStream(resultTable).print("table1");
        //使用table api的方式
        Table resultTable2 = eventTable.select($("url"), $("user"));
        tableEnv.toDataStream(resultTable2).print("table2");
        env.execute(TableAPI_01_Demo.class.getSimpleName());
    }
}
